package com.atguigu.cn.dataStream.api

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
  * @author: yangShen
  * @Description: 测输出流
  *              对于乱序数据，方案一设置watermark进行延时处理
  *                           方案二生成侧输出流
  * @Date: 2020/4/21 13:49 
  */
object SideOutputTest {
  def main(args: Array[String]): Unit = {
    //1.environment
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)   //设置时间语义类型


    // 2.source
    //    val inputStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")
    val inputStream = environment.socketTextStream("10.16.26.16",7777)

    // 3.transform 操作，但没有标记返回类型
    val dataStream = inputStream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
      //设置时间戳
      //处理乱序数据，延迟1秒钟上涨水位的操作(当时间到第11秒时，则认为第10秒为处理时间)
      //EventTime 的时钟设置 = (element.timestamp * 1000) - 1
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
      override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000
    })

    //底层API：ProcessFunction：测输出流
    val processStream = dataStream.process(new FreezingAlert())

//    dataStream.print("input data")
    processStream.print("process data") //主输出流
    //获取侧输出流并打印输出
    processStream.getSideOutput(new OutputTag[String]("freezing alert")).print("freezing alert data")

    environment.execute("window test")

  }

}
//冰点报警，如果小于32F，输出报警信息到测输出流
class FreezingAlert() extends ProcessFunction[SensorReading, SensorReading]{

  lazy val alertOutput: OutputTag[String] = new OutputTag[String]("freezing alert")

  override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {
    //输出到 测输出流
    if (value.temperature < 32.0){
      ctx.output(alertOutput, "freezing alert for " + value.id)
    }else {
      //输出到 主流
      out.collect(value)
    }
  }
}
